# -*- coding: utf-8 -*-
from datetime import timedelta
from time import time
from utils.operators.rich_sql_sensor import RichSqlSensor


label = f"dm_waybill_false_taking_summary_dt_{int(time())}"
timeout = timedelta(hours=1).seconds
doris_jms_dm__dm_waybill_false_taking_summary_dt = RichSqlSensor(
    task_id='doris_jms_dm__dm_waybill_false_taking_summary_dt',
    pool='broker_load_pool',
    email=['suning@jtexpress.com', 'matthew.xiong@jtexpress.com', 'yl_bigdata@yl-scm.com'],
    task_concurrency=1,
    conn_id='new_doris',
    pre_sql=f"""
    TRUNCATE TABLE jms_dm.dm_waybill_false_taking_summary_dt PARTITION (
         p{{{{ execution_date | cst_ds_nodash }}}}
        ,p{{{{ execution_date | date_add(-1) |cst_ds_nodash }}}}
        ,p{{{{ execution_date | date_add(-2) |cst_ds_nodash }}}}
        ,p{{{{ execution_date | date_add(-3) |cst_ds_nodash }}}}
        ,p{{{{ execution_date | date_add(-4) |cst_ds_nodash }}}}
        ,p{{{{ execution_date | date_add(-5) |cst_ds_nodash }}}}
        ,p{{{{ execution_date | date_add(-6) |cst_ds_nodash }}}}
    );
    LOAD LABEL jms_dm.{label} (
        DATA INFILE(
          "hdfs://{{{{ var.value.hadoop_namespace }}}}/dw/hive/jms_dm.db/external/dm_waybill_false_taking_summary_dt/dt={{{{ execution_date | cst_ds }}}}/*",
          "hdfs://{{{{ var.value.hadoop_namespace }}}}/dw/hive/jms_dm.db/external/dm_waybill_false_taking_summary_dt/dt={{{{ execution_date | date_add(-1) |cst_ds }}}}/*",
          "hdfs://{{{{ var.value.hadoop_namespace }}}}/dw/hive/jms_dm.db/external/dm_waybill_false_taking_summary_dt/dt={{{{ execution_date | date_add(-2) |cst_ds }}}}/*",
          "hdfs://{{{{ var.value.hadoop_namespace }}}}/dw/hive/jms_dm.db/external/dm_waybill_false_taking_summary_dt/dt={{{{ execution_date | date_add(-3) |cst_ds }}}}/*",
          "hdfs://{{{{ var.value.hadoop_namespace }}}}/dw/hive/jms_dm.db/external/dm_waybill_false_taking_summary_dt/dt={{{{ execution_date | date_add(-4) |cst_ds }}}}/*",
          "hdfs://{{{{ var.value.hadoop_namespace }}}}/dw/hive/jms_dm.db/external/dm_waybill_false_taking_summary_dt/dt={{{{ execution_date | date_add(-5) |cst_ds }}}}/*",
          "hdfs://{{{{ var.value.hadoop_namespace }}}}/dw/hive/jms_dm.db/external/dm_waybill_false_taking_summary_dt/dt={{{{ execution_date | date_add(-6) |cst_ds }}}}/*"
        )   
        INTO TABLE dm_waybill_false_taking_summary_dt
        FORMAT AS 'PARQUET'(                               
                taking_date                        
                ,receive_network_code               
                ,order_source_code                  
                ,taking_user_code  
                ,customer_code                 
                ,customer_name                      
                ,receive_fran_code                  
                ,receive_agent_code                 
                ,receive_virt_agent_code            
                ,receive_network_name               
                ,receive_fran_name                  
                ,receive_agent_name                 
                ,receive_virt_agent_name            
                ,taking_user_name   
                ,receive_network_taking_cnt   
                ,receive_network_send_cnt     
                ,collect_arr_cnt              
                ,center_arr_cnt                               
                ,receive_network_false_taking_cnt   
                ,receive_network_should_hand_in_cnt 
                ,interrupted_taking_cnt             
                ,false_taking_no_send_cnt           
                ,false_taking_no_hand_in_cnt        
                ,taking_duration_hour                                      
        )
        columns from path as (dt)
        set(
            taking_date                         =    taking_date                        
            ,receive_network_code               =    receive_network_code              
            ,order_source_code                  =    order_source_code                 
            ,taking_user_code                   =    taking_user_code  
            ,customer_code                      =    customer_code                
            ,customer_name                      =    customer_name                     
            ,receive_fran_code                  =    receive_fran_code                 
            ,receive_agent_code                 =    receive_agent_code                
            ,receive_virt_agent_code            =    receive_virt_agent_code           
            ,receive_network_name               =    receive_network_name              
            ,receive_fran_name                  =    receive_fran_name                 
            ,receive_agent_name                 =    receive_agent_name                
            ,receive_virt_agent_name            =    receive_virt_agent_name           
            ,taking_user_name                   =    taking_user_name   
            ,receive_network_taking_cnt         =    receive_network_taking_cnt   
            ,receive_network_send_cnt           =    receive_network_send_cnt     
            ,collect_arr_cnt                    =    collect_arr_cnt              
            ,center_arr_cnt                     =    center_arr_cnt                             
            ,receive_network_false_taking_cnt   =    receive_network_false_taking_cnt  
            ,receive_network_should_hand_in_cnt =    receive_network_should_hand_in_cnt
            ,interrupted_taking_cnt             =    interrupted_taking_cnt            
            ,false_taking_no_send_cnt           =    false_taking_no_send_cnt          
            ,false_taking_no_hand_in_cnt        =    false_taking_no_hand_in_cnt       
            ,taking_duration_hour               =    taking_duration_hour              
            ,dt                                 =    dt
        )
    )
    WITH BROKER '{{{{ var.json.doris_brokers | random_choice }}}}'
    PROPERTIES ('timeout'='{timeout}', 'max_filter_ratio'='0.0')
    """,
    poke_sql=f"SHOW LOAD FROM jms_dm WHERE label = '{label}' ORDER BY CreateTime DESC LIMIT 1",
    sql_on_kill=f"CANCEL LOAD FROM jms_dm WHERE LABEL = '{label}'",
    success=lambda r: r[2] == 'FINISHED',
    failure=lambda r: (r[2] is not None and r[2] == 'CANCELLED', str(r[7])),
    poke_interval=60,
    execution_timeout=timedelta(seconds=timeout + 120), )